package com.zxy.zxyrpc.registry;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import com.zxy.zxyrpc.config.RegistryConfig;
import com.zxy.zxyrpc.exception.RpcException;
import com.zxy.zxyrpc.model.ErrorCode;
import com.zxy.zxyrpc.model.ServiceMetaInfo;
import io.etcd.jetcd.*;
import io.etcd.jetcd.kv.GetResponse;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.options.WatchOption;
import io.etcd.jetcd.watch.WatchEvent;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * ETCD 注册中心（分布式键值存储的王者--ETCD -分布式键值存储的王者--ETCD
 * <a href="https://www.cnblogs.com/guangdelw/p/18689243">...</a>
 *
 * @author zxy
 * @since 2025/5/18
 **/
public class EtcdRegistry implements Registry {
    /**
     * 根节点（业务前缀）
     */
    private static final String ETCD_ROOT_PATH = "/rpc/";
    /**
     * 本机注册节点 key 集合（用于维护心跳续期）
     */
    private final Set<String> localRegisterNodeKeySet = new HashSet<>();
    /**
     * 注册中心服务缓存（提升服务发现的性能）
     */
    private final RegistryServiceCache registryServiceCache = new RegistryServiceCache();
    /**
     * 监听的 key 集合（ConcurrentHashSet 防止并发冲突）
     */
    private final Set<String> watchingKeySet = new ConcurrentHashSet<>();
    private Client client;
    private KV kvClient;

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Client c = Client.builder().endpoints("http://localhost:2379").build();
        KV kv = c.getKVClient();

        ByteSequence key = ByteSequence.from("test_key".getBytes());
        ByteSequence value = ByteSequence.from("test_value".getBytes());

        kv.put(key, value).get();
        CompletableFuture<GetResponse> getFuture = kv.get(key);
        System.out.println(getFuture.get().getKvs().get(0));
        kv.delete(key).get();
    }

    @Override
    public void init(RegistryConfig registryConfig) {
        // address 支持传入多个地址
        client = Client.builder()
                .endpoints(registryConfig.getAddress())
                .connectTimeout(Duration.ofMillis(registryConfig.getTimeout()))
                .build();
        kvClient = client.getKVClient();
        // todo 一旦某个服务使用注册中心就会开启心跳机制是不合理的
        // heartBeat();
    }

    @Override
    public void registry(ServiceMetaInfo serviceMetaInfo) throws Exception {
        // lease 和 Kv 客户端
        Lease leaseClient = client.getLeaseClient();
        // 时间为30秒的租约
        long leaseId = leaseClient.grant(30).get().getID();
        // 设置要存储的键值对
        String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        ByteSequence key = ByteSequence.from(registerKey, StandardCharsets.UTF_8);
        ByteSequence value = ByteSequence.from(JSONUtil.toJsonStr(serviceMetaInfo), StandardCharsets.UTF_8);
        // 键值对与租约关联
        PutOption putOption = PutOption.builder().withLeaseId(leaseId).build();
        kvClient.put(key, value, putOption).get();

        // 添加节点信息到本地缓存
        localRegisterNodeKeySet.add(registerKey);
    }

    @Override
    public void unRegistry(ServiceMetaInfo serviceMetaInfo) {
        String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        kvClient.delete(ByteSequence.from(registerKey, StandardCharsets.UTF_8));
        // 删除缓存
        localRegisterNodeKeySet.remove(registerKey);
    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        // 先走服务缓存
        List<ServiceMetaInfo> serviceMetaInfos = registryServiceCache.readCache(serviceKey);
        if (!CollUtil.isEmpty(serviceMetaInfos)) {
            System.out.println("服务走了緩存");
            return serviceMetaInfos;
        }

        try {
            // 第一次访问时，开启服务目录/前缀监听
            if (!registryServiceCache.containsKey(serviceKey))
                watch(serviceKey, null);
            // 前缀查询
            // 前缀索引，后面一定要加 "/"(?骗我)
            // String searchPrefix = ETCD_ROOT_PATH + serviceKey + "/";
            String searchPrefix = ETCD_ROOT_PATH + serviceKey;
            GetOption getOption = GetOption.builder().isPrefix(true).build();
            List<KeyValue> kvs = kvClient.get(
                            ByteSequence.from(searchPrefix, StandardCharsets.UTF_8),
                            getOption)
                    .get()
                    .getKvs();
            // 解析服务列表信息
            List<ServiceMetaInfo> serviceMetaInfoList = kvs.stream()
                    .map(kv -> {
                        // String key = kv.getKey().toString(StandardCharsets.UTF_8);
                        // watch(serviceKey, key);
                        String value = kv.getValue().toString(StandardCharsets.UTF_8);
                        return JSONUtil.toBean(value, ServiceMetaInfo.class);
                    }).collect(Collectors.toList());
            // 写入服务缓存
            registryServiceCache.writeCache(serviceKey, serviceMetaInfoList);
            return serviceMetaInfoList;
        } catch (Exception e) {
            throw new RpcException(ErrorCode.OPERATION_ERROR, "获取服务列表失败");
        }
    }

    @Override
    public void destroy() {
        System.out.println("当前注册服务节点全部下线");
        // 遍历本服务下，所有的服务节点（我感觉也可以等key过期删除）
        // todo 有点不合理，我不用了，可能别的项目还在用。假设没别的项目用吧
        for (String key : localRegisterNodeKeySet) {
            try {
                kvClient.delete(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
            } catch (Exception e) {
                throw new RpcException(ErrorCode.OPERATION_ERROR, key + "节点下线失败");
            }
        }
        CronUtil.stop();
        // 释放锁（是否需要锁检查啊）
        if (kvClient != null)
            kvClient.close();
        if (client != null)
            client.close();
    }

    /**
     * 注意一次只能开启一个定时任务
     * todo 逻辑有些问题
     */
    @Override
    public void heartBeat() {
        // 10秒续期
        CronUtil.schedule("*/10 * * * * *", (Task) () -> {
            // 遍历所有注册的服务节点。问题：时间统一可能并不好（让我想起了缓存雪崩）
            for (String key : localRegisterNodeKeySet) {
                try {
                    List<KeyValue> kvs = kvClient.get(ByteSequence.from(key, StandardCharsets.UTF_8))
                            .get()
                            .getKvs();
                    // 该服务节点已过期（需要服务启动，并重新注册.被动下线）
                    if (CollUtil.isEmpty(kvs))
                        continue;
                    // 节点未过期，重新注册（续期/刷新过期时间）
                    for (KeyValue kv : kvs) {
                        String value = kv.getValue().toString(StandardCharsets.UTF_8);
                        ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                        registry(serviceMetaInfo);
                    }
                } catch (Exception e) {
                    throw new RpcException(ErrorCode.OPERATION_ERROR, key + " 续签失败");
                }
            }
        });
        // 支持秒级别的定时任务
        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }

    @Override
    public void watch(String serviceKey, String serviceNodeKey) {
        Watch watchClient = client.getWatchClient();
        // 之前未被监听，开启监听
        boolean newWatch = watchingKeySet.add(serviceKey);
        if (newWatch) {
            WatchOption watchOption = WatchOption.builder().isPrefix(true).build();
            watchClient.watch(ByteSequence.from(ETCD_ROOT_PATH + serviceKey, StandardCharsets.UTF_8), watchOption, res -> {
                for (WatchEvent event : res.getEvents()) {
                    // 注册中心的服务节点
                    KeyValue keyValue = event.getKeyValue();
                    String watchServiceNodeKey = keyValue.getKey().toString(StandardCharsets.UTF_8);
                    switch (event.getEventType()) {
                        // key 删除时触发
                        case DELETE:
                            System.out.println(watchServiceNodeKey + " 服务节点过期了!");
                            registryServiceCache.removeServiceNodeCache(serviceKey, watchServiceNodeKey);
                        case PUT:
                            // 可能是心态机制
                            if (registryServiceCache.containsCache(serviceKey, watchServiceNodeKey))
                                break;
                            ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(keyValue.getValue().toString(StandardCharsets.UTF_8), ServiceMetaInfo.class);
                            registryServiceCache.writeNodeCache(serviceKey, serviceMetaInfo);
                        default:
                            break;
                    }
                }
            });
        }
    }
}
